package com.itheima.devops.writeQueue;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
   
/**
   * 队列管理器，负责管理写桶和读桶集合
   */   
public class QueueManager<T> {
    private final int bucketMaxSize;            // 每个桶的最大容量
    private final long refreshIntervalMillis;   // 定时刷新间隔(毫秒)
    private Bucket<T> currentWriteBucket;       // 当前写入桶
    private final Queue<Bucket<T>> readBuckets;  // 只读桶集合(等待消费)
    private final ReentrantLock writeLock;      // 写操作锁
    private final ReentrantLock refreshLock;    // 刷新操作锁
    private ScheduledExecutorService scheduler; // 定时任务调度器
    private final Semaphore semaphore = new Semaphore(5);
    private final ExecutorService executor=Executors.newFixedThreadPool(8);


    public QueueManager(int bucketMaxSize, long refreshIntervalMillis) {
        this.bucketMaxSize = bucketMaxSize;
        this.refreshIntervalMillis = refreshIntervalMillis;
        this.currentWriteBucket = new Bucket<>(bucketMaxSize);
        this.readBuckets = new ConcurrentLinkedQueue<>();
        this.writeLock = new ReentrantLock();
        this.refreshLock = new ReentrantLock();
        // 创建消费线程池
        startScheduledRefresh();
    }

    // 添加数据到当前写桶
    public void addData(T item) {
        writeLock.lock();
        try {
            // 如果当前写桶已满，创建新桶
            if (!currentWriteBucket.addData(item)) {
                refreshCurrentBucket();
                currentWriteBucket.addData(item);
            }
        } finally {
            writeLock.unlock();
        }
    }

    // 定时刷新当前写桶到读桶集合
    private void startScheduledRefresh() {
        scheduler = Executors.newSingleThreadScheduledExecutor();
//        scheduler.scheduleAtFixedRate(() -> {
//            try {
//                refreshCurrentBucket();
//                refreshLock.lock();
//                try {
//                    consumeMultipleBuckets(Integer.MAX_VALUE);
//                } finally {
//                    refreshLock.unlock();
//                }
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//        }, refreshIntervalMillis, refreshIntervalMillis, TimeUnit.MILLISECONDS);

        scheduler.scheduleAtFixedRate(() -> {
            System.out.println("定时任务开始执行");
            try {
                refreshCurrentBucket();
                refreshLock.lock();
                try {
                    consumeMultipleBuckets(Integer.MAX_VALUE);
                } finally {
                    refreshLock.unlock();
                }
            } catch (Exception e) {
                // 可选择重启线程或报警
            }
        }, 0, 10, TimeUnit.SECONDS);
    }




    // 刷新当前写桶到读桶集合
    private void refreshCurrentBucket() {
        refreshLock.lock();
        try {
            writeLock.lock();
            try {
                if (currentWriteBucket.size() > 0 && currentWriteBucket.markAsReadOnly()) {
                    readBuckets.offer(currentWriteBucket);
                    currentWriteBucket = new Bucket<>(bucketMaxSize);
                }
            } finally {
                writeLock.unlock();
            }
        } finally {
            refreshLock.unlock();
        }
    }



    public boolean consumeMultipleBuckets(int maxBuckets) {
        int consumedCount = 0;
        long startTime = System.currentTimeMillis();
        System.out.println(startTime + "-->本次待消费数据桶个数为：" + readBuckets.size());
        while (!readBuckets.isEmpty()){
            Bucket<T> bucket = readBuckets.poll();
            try {
                semaphore.acquire(); // 获取许可（阻塞直到有空闲资源）
                Future<String> future = executor.submit(() -> {
                    try {
                        dataConsumed(bucket.consumeData());
                        return "1";
                    } finally {
                        semaphore.release(); // 释放许可
                    }
                });
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(startTime + "-->Consumed single bucket: " + bucket.size()
                    + " items, 消费延迟: " + (startTime - bucket.getCreateTimestamp()) + "ms");
        }
        System.out.println("本次实际消费桶数量: " + consumedCount);
        return consumedCount > 0;
    }

    private void dataConsumed(List<T> data) {
        System.out.println("开始消费数据: " + data.size() + "条");
        System.out.println(data);
        try {
            // 模拟处理时间
            TimeUnit.MILLISECONDS.sleep(100); // 改为更短的模拟时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("数据消费完成");
    }

}
